package ricelee.demo.rxjava;

import android.os.Bundle;
import android.support.annotation.Nullable;
import android.support.v7.app.AppCompatActivity;
import android.support.v7.widget.Toolbar;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.TextView;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

//操作符
public class OperatorActivity extends AppCompatActivity implements View.OnClickListener {

    private Button mapBtn;
    private Button flatMapBtn;
    private Button concatMapBtn;
    private Button zipBtn;

    private TextView sendTv;
    private TextView receiveTv;
    private StringBuilder sendStringBuilder;
    private StringBuilder receiveStringBuilder;


    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_operator);
        setSupportActionBar((Toolbar) findViewById(R.id.toolbar));
        getSupportActionBar().setTitle("操作符");

        initView();
    }

    private void initView() {
        mapBtn = (Button) findViewById(R.id.btn_operator_map);
        mapBtn.setOnClickListener(this);

        flatMapBtn = (Button) findViewById(R.id.btn_operator_flatmap);
        flatMapBtn.setOnClickListener(this);

        concatMapBtn = (Button) findViewById(R.id.btn_operator_concatMap);
        concatMapBtn.setOnClickListener(this);

        zipBtn = (Button) findViewById(R.id.btn_operator_zip);
        zipBtn.setOnClickListener(this);

        sendTv = (TextView) findViewById(R.id.tv_send);
        receiveTv = (TextView) findViewById(R.id.tv_receive);


    }

    private void initData() {
        sendStringBuilder = null;
        receiveStringBuilder = null;
        sendStringBuilder = new StringBuilder("发出的数据:\n");
        receiveStringBuilder = new StringBuilder("收到的数据:\n");
    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.btn_operator_map:
                map();
                break;
            case R.id.btn_operator_flatmap:
                flatMap();
                break;
            case R.id.btn_operator_concatMap:
                concatMap();
                break;
            case R.id.btn_operator_zip:
                zip();
                break;

        }
    }


    private void map() {
        initData();
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                sendStringBuilder.append(1 + "\n");
                emitter.onNext(2);
                sendStringBuilder.append(2 + "\n");
                emitter.onNext(3);
                sendStringBuilder.append(3 + "\n");
                sendTv.setText(sendStringBuilder.toString());
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(@NonNull Integer integer) throws Exception {
                return "This is result " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                receiveStringBuilder.append(s + "\n");
                receiveTv.setText(receiveStringBuilder);
            }
        });

    }

    private void flatMap() {
        initData();
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                sendStringBuilder.append("1\n");
                emitter.onNext(2);
                sendStringBuilder.append("2\n");
                emitter.onNext(3);
                sendStringBuilder.append("3\n");
                sendTv.setText(sendStringBuilder.toString());
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                return Observable.fromIterable(list).delay(100, TimeUnit.MILLISECONDS);

            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                receiveStringBuilder.append(s + "\n");
//                receiveTv.setText(receiveStringBuilder.toString());
// --报错io.reactivex.exceptions.OnErrorNotImplementedException
                Log.i(OperatorActivity.class.getSimpleName(), receiveStringBuilder.toString());
            }
        });
    }

    private void concatMap() {
        initData();
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                sendStringBuilder.append(1 + "\n");
                emitter.onNext(2);
                sendStringBuilder.append(2 + "\n");
                emitter.onNext(3);
                sendStringBuilder.append(3 + "\n");
                sendTv.setText(sendStringBuilder.toString());
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                return Observable.fromIterable(list).delay(100, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                receiveStringBuilder.append(s + "\n");
//                receiveTv.setText(receiveStringBuilder.toString());
// --报错io.reactivex.exceptions.OnErrorNotImplementedException
                Log.i(OperatorActivity.class.getSimpleName(), receiveStringBuilder.toString());
            }
        });
    }

    private void zip() {
        initData();
        Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                sendStringBuilder.append("Integer emitter:" + 1 + "\n");
                emitter.onNext(2);
                sendStringBuilder.append("Integer emitter:" + 2 + "\n");
                emitter.onNext(3);
                sendStringBuilder.append("Integer emitter:" + 3 + "\n");
                sendTv.setText(sendStringBuilder.toString());
            }
        });
        Observable<String> stringObservable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("A");
                sendStringBuilder.append("String emitter:" + "A\n");
                emitter.onNext("B");
                sendStringBuilder.append("String emitter:" + "B\n");
                emitter.onNext("C");
                sendStringBuilder.append("String emitter:" + "C\n");
                sendTv.setText(sendStringBuilder.toString());
            }
        });

        Observable.zip(integerObservable, stringObservable, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(@NonNull Integer integer, @NonNull String s) throws Exception {
                return integer + s;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                receiveStringBuilder.append("onSubscribe\n");
                receiveTv.setText(receiveStringBuilder.toString());
            }

            @Override
            public void onNext(@NonNull String s) {
                receiveStringBuilder.append("onNext:" + s+"\n");
                receiveTv.setText(receiveStringBuilder.toString());
            }

            @Override
            public void onError(@NonNull Throwable e) {
                receiveStringBuilder.append("onError\n");
                receiveTv.setText(receiveStringBuilder.toString());

            }

            @Override
            public void onComplete() {
                receiveStringBuilder.append("onComplete\n");
                receiveTv.setText(receiveStringBuilder.toString());
            }
        });


    }
}
